package com.oracle.calllog.etl.runner;

import com.oracle.calllog.etl.util.HBaseConnectionUtil;
import com.oracle.calllog.etl.util.HBaseUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import com.oracle.calllog.etl.util.Constants;
import com.oracle.calllog.etl.mapper.CalllogsEtl2HBaseMapper;
import com.oracle.calllog.etl.reducer.CalllogsEtl2HBaseReducer;
import org.apache.hadoop.util.ToolRunner;


/******************************************
 * @ClassName CalllogsEtl2HBaseReducer
 * @Description ETL流程的Runner提交工具类
 * @Author Mr.Chen
 * @Date 2022/6/11 21:13
 * @Version 1.0
 *****************************************/

public class CalllogsEtl2HBaseRunner implements Tool {
    private Configuration conf;

    @Override
    public int run(String[] args) throws Exception {
        // 获取job信息
        Job job = Job.getInstance(this.conf);

        // 设置Job作业
        job.setJarByClass(CalllogsEtl2HBaseRunner.class);

        // map阶段的Job作业设置
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Put.class);

        job.setMapperClass(CalllogsEtl2HBaseMapper.class);

        FileInputFormat.setInputPaths(job, new Path(args[0]));

        // Reduce输出数据的目标Table的表名从属性文件中获取
        String tableName = Constants.TABLE_NAME;
        //如果属性文件中的表名包含命名空间，则先初始化命名空间
        String[] tableNameParts = Constants.TABLE_NAME.split(":");
        if (tableNameParts.length>1){
            //如果命名空间不存在，则先初始化命名空间
            if(!HBaseUtil.isExistNamespace(this.conf, tableNameParts[0])){
                HBaseUtil.initNamespace(this.conf, tableNameParts[0]);
            }
        }
        //如果表格不存在，则先以预分区的方式创建表格
        if (!HBaseUtil.isExistTable(this.conf, Constants.TABLE_NAME)){
            //以预分区的方式创建表格
            HBaseUtil.createTable(this.conf, Constants.TABLE_NAME, Constants.REGION_NUMS, Constants.FAMILY_CALL);
        }
        HBaseConnectionUtil.closeConnection();

        //初始化Job作业中Reduce阶段的设置
        TableMapReduceUtil.initTableReducerJob(tableName, CalllogsEtl2HBaseReducer.class, job);

        // 提交
        boolean result = job.waitForCompletion(true);
        return result ? 0 : 1;
    }

    @Override
    public void setConf(Configuration configuration) {
        this.conf = HBaseConfiguration.create(configuration);
    }

    @Override
    public Configuration getConf() {
        return conf;
    }

    public static void main(String[] args) {
        if (args == null || args.length < 1) {
            System.out.println("no arguments!please run with input path!");
            return;
        }

        try {
            ToolRunner.run(new CalllogsEtl2HBaseRunner(), args);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
